package cn._51doit.flink.day06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * 将ValueState的底层实现
 *
 * Flink的State分为两种：KeyedState(KeyBy之后对应的State)，和OperatorState（没有keyBy的State）
 *
 * ValueState是KeyedState中的一种
 *
 * 1.KeyedState底层是一个Map结构
 * 2.如果想要容错，必须开启checkpointing，并且按照Flink的状态编程API进行编程（将中间结果保存都Flink特殊的变量中）
 *
 * ValueState : Map<Key, Value>
 * MapState   : Map<Key, Map<k, v>>
 * ListState  : Map<Key, List<v>>
 *
 */
public class ListStateDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //开启checkpoint
        env.enableCheckpointing(5000);

        //将同一个用户，最近的10个行为保存起来
        //u001,view
        //u001,pay
        //u002,view
        //u002,view
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888, "\n", 5);

        SingleOutputStreamOperator<Tuple2<String, String>> tpStream = lines.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String value) throws Exception {
                if (value.startsWith("error")) {
                    throw new RuntimeException("数据出问题了！");
                }
                String[] fields = value.split(",");
                String uid = fields[0];
                String event = fields[1];
                return Tuple2.of(uid, event);
            }
        });

        KeyedStream<Tuple2<String, String>, String> keyedStream = tpStream.keyBy(t -> t.f0);

        //将用一个用户的行为数据按照先后顺序保存起来
        SingleOutputStreamOperator<Tuple2<String, List<String>>> res = keyedStream.map(new UserEventFunction());

        res.print();

        env.execute();


    }

    private static class UserEventFunction extends RichMapFunction<Tuple2<String, String>, Tuple2<String, List<String>>> {

        private ListState<String> listState;

        @Override
        public void open(Configuration parameters) throws Exception {
            //定义状态描述器
            ListStateDescriptor<String> stateDescriptor = new ListStateDescriptor<>("event-state", String.class);
            //初始化或恢复状态
            listState = getRuntimeContext().getListState(stateDescriptor);
        }

        @Override
        public Tuple2<String, List<String>> map(Tuple2<String, String> input) throws Exception {
            String event = input.f1;
            listState.add(event);
            ArrayList<String> events = (ArrayList<String>) listState.get();
            if (events.size() > 10) {
                events.remove(0);
            }
            return Tuple2.of(input.f0, events);
        }
    }

}
